package com.qing;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.Observer;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Consumer;
import io.reactivex.schedulers.Schedulers;

public class HelloRx {
	public static void main(String[] args) {
		//可观察者
		Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() {
			@Override
			public void subscribe(ObservableEmitter<String> observableEmitter) throws Exception {
				observableEmitter.onNext("hello");
				observableEmitter.onNext("www.baidu.com");
				observableEmitter.onNext("你好！小纳");
			}
		});
		//消费者
		Consumer<String> consumer = new Consumer<String>() {
			@Override
			public void accept(String s) throws Exception {
				System.out.println(Thread.currentThread().getName() + " === consumer=== " + s);
			}
		};


//        observable.subscribe(consumer);
		//实现异步
		observable.observeOn(Schedulers.newThread()).subscribe(consumer);

		//观察者
		Observer<String> observer = new Observer<String>() {
			@Override
			public void onSubscribe(Disposable disposable) {

			}

			@Override
			public void onNext(String s) {
				System.out.println(Thread.currentThread().getName() + " ======observer====== " + s);
			}

			@Override
			public void onError(Throwable throwable) {

			}

			@Override
			public void onComplete() {

			}
		};

//        observable.subscribe(observer);
		observable.observeOn(Schedulers.newThread()).subscribe(observer);
		for (; ; ) ; //死循环阻塞
	}
}
